// Copyright (c) 2019 Presto Labs Pte. Ltd.
// Author: jshin

#include "coin2/exchange/base/feed/processor.h"

#include <experimental/filesystem>
#include <memory>
#include <set>
#include <stdexcept>
#include <string>
#include <utility>
#include <vector>

#include "coin2/exchange/base/api_util/json_schema.h"
#include "coin2/exchange/base/executor/topic_distributor.h"
#include "coin2/exchange/base/feed/distributor.h"
#include "coin2/exchange/base/feed/subscriber.h"
#include "coin2/exchange/base/feed/update_manager.h"
#include "coin2/exchange/base/feed/ws/feed_subscription_spec.h"
#include "coin2/exchange/base/market/market_exchange_api.h"
#include "coin2/exchange/di/constant.h"
#include "coin2/flow/topic_subscription.h"

namespace coin2::exchange::base::feed {
namespace impl {

FeedParsingProcessor::FeedParsingProcessor(
    const std::string& system_name,
    const MarketExchangeApi& mea,
    const di::ConstantInjection& ci,
    const RawParserConfig& parser_config,
    const std::string& schema_path)
    : parser_config_(parser_config),
      upd_mgr_(
          system_name,
          mea,
          parser_config_.enable_diff_book(),
          parser_config_.enable_telemetry()),
      subscriber_(nullptr) {
  const auto& ch_map = ci.GetFeedSubscriptionSpec(mea).channels;
  for (const auto& ch_name : parser_config_.channels()) {
    if (ch_map.find(ch_name) == ch_map.end()) {
      throw std::runtime_error(
          "[FeedParsingProcessor] channel " + ch_name + " not found in " + mea.String());
    }
    // for rest api channels, chnnel request only contains parameters,
    // can't use it as signification of channel
    // add channel name to identify channels
    channel_strings_.insert(std::make_pair(ch_name, ch_map.at(ch_name)));
  }

  if (!schema_path.empty()) {
    schema_.reset(new JsonSchema(schema_path));
  }
}

FeedParsingProcessor::~FeedParsingProcessor() = default;

}  // namespace impl
}  // namespace coin2::exchange::base::feed
